package com.jch.learn.demo.code;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

// 固定文件大小 控制读写位置 循环读写队列映射文件
public class SimpleLocalMessageQueue {

    /**
     * 当前数据文件大小
     */
    private final AtomicLong currentDataFileSpaceUsage;

    /**
     * 数据文件的容量上限
     */
    private final long maxDataFileSize;

    /**
     * 数据文件缓冲区
     */
    private final MappedByteBuffer dataBuf;

    /**
     * 元数据文件缓冲区
     */
    private final MappedByteBuffer metaBuf;

    /**
     * 队列元数据文件大小 （head + tail）
     */
    private static final int META_SIZE = 8;

    /**
     * 通道锁
     */
    final ReentrantLock lock = new ReentrantLock();

   /**
    //非空条件
    private final Condition notEmpty = lock.newCondition();
    //队列已满条件
    private final Condition notFull = lock.newCondition();
     */

    /**
     * 简单本地消息队列
     * @param path 队列映射文件路径
     * @param queueName 队列名称（队列文件名称）
     * @throws IOException
     */
    public SimpleLocalMessageQueue(String path, String queueName) throws IOException {
        this(path, queueName, 10 * 1024 * 1024);
    }

    /**
     * 简单本地消息队列
     *
     * @param path            队列映射文件路径
     * @param queueName       队列名称（队列文件名称）【.data数据文件，.meta元数据文件】
     * @param maxDataFileSize 数据文件大小
     */
    public SimpleLocalMessageQueue(String path, String queueName, long maxDataFileSize) throws IOException {
        if (!path.endsWith(File.separator)) {
            path += File.separator;
        }
        this.maxDataFileSize = maxDataFileSize;
        RandomAccessFile dataFile = null;
        RandomAccessFile metaFile = null;
        try {
            // 若不存在文件 则会自动创建
            dataFile = new RandomAccessFile(path + queueName + ".data", "rw");
            metaFile = new RandomAccessFile(path + queueName + ".meta", "rw");
            dataBuf = dataFile.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, maxDataFileSize);
            metaBuf = metaFile.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, META_SIZE);
        } finally {
            if (dataFile != null) {
                dataFile.close();
            }
            if (metaFile != null) {
                metaFile.close();
            }
        }
        // 初始化 当前数据文件实际使用空间大小
        if (tail() >= head()) {
            currentDataFileSpaceUsage = new AtomicLong(tail() - head());
        } else {
            currentDataFileSpaceUsage = new AtomicLong(maxDataFileSize - head() + tail());
        }
    }


    /**
     * 获取队列头消息位置
     *
     * @return 头位置
     */
    private int head() {
        int head = metaBuf.getInt(0);
        return head;
    }

    /**
     * 设置新的头消息位置
     *
     * @param newHead 新位置
     */
    private void head(int newHead) {
        metaBuf.putInt(0, newHead);
    }

    /**
     * 获取队列尾消息位置
     *
     * @return 尾位置
     */
    private int tail() {
        int tail = metaBuf.getInt(4);
        return tail;
    }

    /**
     * 设置新的尾消息位置
     *
     * @param newTail 新位置
     */
    private void tail(int newTail) {
        metaBuf.putInt(4, newTail);
    }

    /**
     * 入队
     *
     * @param data 消息
     */
    public void enqueue(byte[] data) {
//        if ((currentDataFileSpaceUsage.get() + data.length + 4) > maxDataFileSize) {
//            throw new IllegalArgumentException("Beyond the maximum !");
//        }

        lock.lock();
        if (currentDataFileSpaceUsage.get() >= maxDataFileSize) {
            lock.unlock();
            throw new IllegalStateException("queue is full！");
        }
        try {
            int tail = tail();
            System.out.println("old tail position:" + tail);
            int length = data.length + 4;
            if (tail + length >= maxDataFileSize){
                if (head() > length){ // 重置文件尾指针
                    System.out.println("重置尾指针");
                    tail = 0;
                } else {
                    throw new IllegalArgumentException("queue space is not enough!");
                }
            }
            dataBuf.position(tail);
            dataBuf.putInt(data.length); // 消息长度
            dataBuf.put(data);
            System.out.println("msg data length: " + data.length);
            currentDataFileSpaceUsage.addAndGet(length);
            int newTailPosition = tail + length;
            if (newTailPosition == maxDataFileSize){ // 正好占满空间 则将尾指针置为0
                newTailPosition = 0;
            }
            dataBuf.force();//刷新到本地文件
            tail(newTailPosition); //设置新尾位置
            System.out.println("new tail position:" + newTailPosition);
            System.out.println("currentDataFileSpaceUsage:" + currentDataFileSpaceUsage);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 出队
     *
     * @return 消息byte[]
     */
    public byte[] dequeue() {
//        if (tail() == head()) {
//            return null;
//        }

        byte[] data = null;
        lock.lock();
        if (currentDataFileSpaceUsage.get() == 0) {
            lock.unlock();
            return null;
        }
        try {
            int head = head();
            System.out.println("old head position:" + head);
            dataBuf.position(head);
            int length = dataBuf.getInt();
            System.out.println("get msg data length:" + length);
            data = new byte[length];
            dataBuf.get(data);
            // 新头消息位置 和 已消费消息占用空间大小
            int newHeadPosition = head + length + 4;
            currentDataFileSpaceUsage.set(currentDataFileSpaceUsage.get() - length - 4); // 更新当前文件大小

            // 更新头位置
            if (maxDataFileSize - newHeadPosition == 0) {// 剩余文件容量
                newHeadPosition = 0;
            } else {
                dataBuf.position(newHeadPosition);
                int nextMsgLength = dataBuf.getInt();
                if (nextMsgLength == 0 && tail() < head){
                    newHeadPosition = 0;
                }
            }

            head(newHeadPosition);
            System.out.println("new head position:" + newHeadPosition);
            System.out.println("currentDataFileSpaceUsage:" + currentDataFileSpaceUsage);
        } finally {
            lock.unlock();
        }
        return data;
    }

}
